package com.atguigu.dw.gmall.realtime.util


import kafka.serializer.StringDecoder
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.KafkaUtils

/**
 * description ：读取kafka数据工具类
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/11 18:12
 * modified By ：
 * version:    : 1.0
 */
object MyKafkaUtil {


  def getKafkaStream(ssc: StreamingContext, topics: Set[String]) = {

    // 构建参数
    val propertiesUtil = PropertiesUtil("config.properties")
    val param = Map[String, String](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> propertiesUtil.getProperty("kafka.broker.list"),
      ConsumerConfig.GROUP_ID_CONFIG -> propertiesUtil.getProperty("kafka.group")
    )

    // 读取kafka数据
    val dawStream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, param, topics)

    // 将结果返回
    dawStream
  }

}
